package com.ityuwen.mengyagushi.album.service.impl;

import com.ityuwen.mengyagushi.album.mapper.AlbumAttributeValueMapper;
import com.ityuwen.mengyagushi.album.mapper.AlbumInfoMapper;
import com.ityuwen.mengyagushi.album.mapper.AlbumStatMapper;
import com.ityuwen.mengyagushi.album.service.AlbumInfoService;
import com.ityuwen.mengyagushi.common.cache.GuiGuCache;
import com.ityuwen.mengyagushi.common.constant.KafkaConstant;
import com.ityuwen.mengyagushi.common.constant.RedisConstant;
import com.ityuwen.mengyagushi.common.constant.SystemConstant;
import com.ityuwen.mengyagushi.common.service.KafkaService;
import com.ityuwen.mengyagushi.common.util.AuthContextHolder;
import com.ityuwen.mengyagushi.model.album.AlbumAttributeValue;
import com.ityuwen.mengyagushi.model.album.AlbumInfo;
import com.ityuwen.mengyagushi.model.album.AlbumStat;
import com.ityuwen.mengyagushi.query.album.AlbumInfoQuery;
import com.ityuwen.mengyagushi.vo.album.AlbumAttributeValueVo;
import com.ityuwen.mengyagushi.vo.album.AlbumInfoVo;
import com.ityuwen.mengyagushi.vo.album.AlbumListVo;
import com.ityuwen.mengyagushi.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.base.Charsets;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.*;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumAttributeValueMapper attributeValueMapper;

    @Autowired
    private AlbumStatMapper statMapper;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private RedisTemplate redisTemplate;

//    @Autowired
//    private RedisDistributedClient redisDistributedClient;

    @Autowired
    private RedissonClient redissonClient;

    @Transactional
    @Override
    public void saveAlbumInfo(AlbumInfo albumInfo) {
        // 1.保存专辑信息表：album_info
        Long userId = AuthContextHolder.getUserId();
        albumInfo.setUserId(userId == null ? 1 : userId); // 判断用户id是否为空，如果为空则取默认值1
        albumInfo.setIncludeTrackCount(0); // 手动设置包含声音数，防止误传
        albumInfo.setIsFinished(null);
        albumInfo.setQualityScore(null);
        if (!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(albumInfo.getPayType())) {
            albumInfo.setTracksForFree(5); // 如果不是免费的专辑则默认前五集免费试听
        }
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
        this.save(albumInfo);
        Long albumId = albumInfo.getId(); // 主键回写获取专辑id

        // 2.保存专辑标签值表：album_attribute_value
        List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();
        if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
            // 遍历标签保存到专辑标签值表
            albumAttributeValueVoList.forEach(albumAttributeValue -> {
                albumAttributeValue.setAlbumId(albumId);
                this.attributeValueMapper.insert(albumAttributeValue);
            });
        }

        // 3.保存专辑统计表：album_stat
        initAlbumStat(albumId, SystemConstant.ALBUM_STAT_PLAY);
        initAlbumStat(albumId, SystemConstant.ALBUM_STAT_SUBSCRIBE);
        initAlbumStat(albumId, SystemConstant.ALBUM_STAT_BROWSE);
        initAlbumStat(albumId, SystemConstant.ALBUM_STAT_COMMENT);

        // 发送消息，让其他服务完成数据同步
        if ("1".equals(albumInfo.getIsOpen())) {
            this.kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        }

        RBloomFilter<String> bloomFilter = this.redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(RedisConstant.ALBUM_INFO_PREFIX + albumId);
    }

    @Override
    public Page<AlbumListVo> findUserAlbumPage(Integer page, Integer limit, AlbumInfoQuery query) {
        // 当前应查询的是当前用户的专辑信息
        Long userId = AuthContextHolder.getUserId();
        query.setUserId(userId != null ? userId : 1L);
        return this.albumInfoMapper.findUserAlbumPage(new Page<AlbumListVo>(page, limit), query);
    }

    @GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX,
            lock = RedisConstant.ALBUM_LOCK_PREFIX,
            timeout = RedisConstant.CACHE_TIMEOUT, random = RedisConstant.ALBUM_TIMEOUT,
            bf = RedisConstant.ALBUM_BLOOM_FILTER
    )
    @Override
    public AlbumInfo getAlbumInfo(Long albumId) {
        // 查询专辑表
        AlbumInfo albumInfo = this.getById(albumId);
        if (albumInfo != null) {
            // 查询专辑标签值表
            List<AlbumAttributeValue> albumAttributeValues = this.attributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>()
                    .eq(AlbumAttributeValue::getAlbumId, albumId));
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValues);
        }
        return albumInfo;
    }

    public AlbumInfo getAlbumInfo2(Long albumId) {

        // 先查询缓存，如果命中则直接返回
        AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(RedisConstant.ALBUM_INFO_PREFIX + albumId);
        if (albumInfo != null) {
            return albumInfo;
        }

        // 添加分布式锁，防止缓存击穿
        RLock fairLock = this.redissonClient.getFairLock(RedisConstant.ALBUM_LOCK_PREFIX + albumId);
        fairLock.lock();

        try {
            // 再查缓存，如果命中则直接返回。因为获取锁的过程中，可能有其他请求把数据放入了缓存
            albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(RedisConstant.ALBUM_INFO_PREFIX + albumId);
            if (albumInfo != null) {
                return albumInfo;
            }

            // 查询专辑表
            albumInfo = this.getById(albumId);
            if (albumInfo != null) {
                // 查询专辑标签值表
                List<AlbumAttributeValue> albumAttributeValues = this.attributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>()
                        .eq(AlbumAttributeValue::getAlbumId, albumId));
                albumInfo.setAlbumAttributeValueVoList(albumAttributeValues);
            }

            // 放入缓存，添加随机值防止缓存雪崩
            this.redisTemplate.opsForValue().set(RedisConstant.ALBUM_INFO_PREFIX + albumId, albumInfo, RedisConstant.CACHE_TIMEOUT + new Random().nextInt(5 * 3600), TimeUnit.SECONDS);

            return albumInfo;
        } finally {
            fairLock.unlock();
        }
    }

    @Override
    public void removeAlbumInfo(Long albumId) {
        // 1.删除专辑信息
        this.removeById(albumId);

        // 2.删除标签信息
        this.attributeValueMapper.delete(new LambdaUpdateWrapper<AlbumAttributeValue>()
                .eq(AlbumAttributeValue::getAlbumId, albumId));

        // 3.删除统计信息
        this.statMapper.delete(new LambdaUpdateWrapper<AlbumStat>().eq(AlbumStat::getAlbumId, albumId));

        this.kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, albumId.toString());
    }

    @Override
    public List<AlbumInfo> findUserAllAlbumList() {
        Long userId = AuthContextHolder.getUserId();
        userId = userId == null ? 1L : userId;
        return this.list(new LambdaQueryWrapper<AlbumInfo>().eq(AlbumInfo::getUserId, userId)
                .select(AlbumInfo::getId, AlbumInfo::getAlbumTitle).orderByDesc(AlbumInfo::getId));
    }

    @Transactional
    @Override
    public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {
        // 1.更新专辑信息
        AlbumInfo albumInfo = new AlbumInfo();
        BeanUtils.copyProperties(albumInfoVo, albumInfo);
        albumInfo.setId(albumId);
        this.updateById(albumInfo);

        // 2.更新专辑标签值表
        this.attributeValueMapper.delete(new LambdaUpdateWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
            albumAttributeValueVoList.forEach(albumAttributeValueVo -> {
                AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
                albumAttributeValue.setAlbumId(albumId);
                this.attributeValueMapper.insert(albumAttributeValue);
            });
        }
        if ("1".equals(albumInfoVo.getIsOpen())) {
            this.kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        }
    }

    @Override
    public Page<AlbumInfo> findAlbumPage(Integer page, Integer limit) {
        return this.page(new Page<>(page, limit));
    }

    @Override
    public List<AlbumAttributeValue> findAttributeValuesByAlbumId(Long albumId) {
        return this.attributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
    }

    @Override
    public AlbumStatVo findAlbumStatByAlbumId(Long albumId) {
        return this.albumInfoMapper.findAlbumStatByAlbumId(albumId);
    }

    @Override
    public void testLock() {
//        lock = new DistributedLock(redisTemplate, "lock");
//        DistributedLock lock = this.redisDistributedClient.getLock("lock");
        RLock lock = this.redissonClient.getFairLock("lock");
        lock.lock();
        try {
            Object num = this.redisTemplate.opsForValue().get("num");
            if (num == null) {
                this.redisTemplate.opsForValue().set("num", 1);
                return;
            }
            Integer count = Integer.valueOf(num.toString());
            this.redisTemplate.opsForValue().set("num", ++count);

//            try {
//                TimeUnit.SECONDS.sleep(1000);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
            this.testSubLock();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void testRead() {
        RReadWriteLock rwLock = this.redissonClient.getReadWriteLock("rwLock");
        rwLock.readLock().lock(10, TimeUnit.SECONDS);
//        rwLock.readLock().unlock();
    }

    @Override
    public void testWrite() {
        RReadWriteLock rwLock = this.redissonClient.getReadWriteLock("rwLock");
        rwLock.writeLock().lock(10, TimeUnit.SECONDS);
    }

    @Override
    public void testLatch() {
        try {
            RCountDownLatch cdl = this.redissonClient.getCountDownLatch("cdl");
            cdl.trySetCount(6);

            cdl.await();
            // 后续锁门操作。。。。
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void testCount() {
        RCountDownLatch cdl = this.redissonClient.getCountDownLatch("cdl");
        // 同学出门操作。。。。
        cdl.countDown();
    }

    @Override
    public List<Long> getChangedAlbumIds(String startTime, String endTime) {

        List<AlbumStat> albumStats = this.statMapper.selectList(new LambdaQueryWrapper<AlbumStat>()
                .gt(AlbumStat::getUpdateTime, startTime)
                .le(AlbumStat::getUpdateTime, endTime)
                .select(AlbumStat::getAlbumId));
        if (CollectionUtils.isEmpty(albumStats)){
            return null;
        }

        return albumStats.stream().map(AlbumStat::getAlbumId).distinct().collect(Collectors.toList());
    }

    @Override
    public List<AlbumStatVo> getAlbumStatsByAlbumIds(List<Long> albumIds) {
        return this.albumInfoMapper.getAlbumStatsByAlbumIds(albumIds);
    }

    public void testSubLock() {
//        DistributedLock lock = this.redisDistributedClient.getLock("lock");
        RLock lock = this.redissonClient.getLock("lock");
        lock.lock();
        System.out.println("================");
        lock.unlock();
    }

    public void testLock2() {
        // 加锁
        String uuid = UUID.randomUUID().toString();
        Boolean lock = this.redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3, TimeUnit.SECONDS);
        if (!lock) {
            try {
                // 睡一会儿重试
                Thread.sleep(40);
                this.testLock();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        } else {
            // 设置过期时间
            // this.redisTemplate.expire("lock", 3, TimeUnit.SECONDS);

            // 执行业务
            Object num = this.redisTemplate.opsForValue().get("num");
            if (num == null) {
                this.redisTemplate.opsForValue().set("num", 1);
                return;
            }
            Integer count = Integer.valueOf(num.toString());
            this.redisTemplate.opsForValue().set("num", ++count);

            // 释放锁
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
                    "then " +
                    "   return redis.call('del', KEYS[1]) " +
                    "else " +
                    "   return 0 " +
                    "end";
            this.redisTemplate.execute(new DefaultRedisScript(script, Boolean.class), Arrays.asList("lock"), uuid);
//            if (StringUtils.equals((String)this.redisTemplate.opsForValue().get("lock"), uuid)){
//                this.redisTemplate.delete("lock");
//            }
        }
    }

    private void initAlbumStat(Long albumId, String statType) {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(0);
        this.statMapper.insert(albumStat);
    }

    public static void main(String[] args) {
//        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
//        System.out.println("定时任务的初始时间：" + System.currentTimeMillis());
//        scheduledExecutorService.scheduleAtFixedRate(() -> {
//            System.out.println("定时任务的执行时间：" + System.currentTimeMillis());
//        }, 5, 10, TimeUnit.SECONDS);

//        System.out.println("定时任务的初始时间：" + System.currentTimeMillis());
//        new Timer().schedule(new TimerTask() {
//            @Override
//            public void run() {
//                System.out.println("定时任务的执行时间：" + System.currentTimeMillis());
//            }
//        }, 5000);

        BloomFilter<String> bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), 20, 0.3);
        bloomFilter.put("1");
        bloomFilter.put("2");
        bloomFilter.put("3");
        bloomFilter.put("4");
        bloomFilter.put("5");
        bloomFilter.put("6");
        bloomFilter.put("7");
        System.out.println(bloomFilter.mightContain("1"));
        System.out.println(bloomFilter.mightContain("3"));
        System.out.println(bloomFilter.mightContain("5"));
        System.out.println(bloomFilter.mightContain("7"));
        System.out.println(bloomFilter.mightContain("8"));
        System.out.println(bloomFilter.mightContain("9"));
        System.out.println(bloomFilter.mightContain("10"));
        System.out.println(bloomFilter.mightContain("11"));
        System.out.println(bloomFilter.mightContain("12"));
        System.out.println(bloomFilter.mightContain("13"));
        System.out.println(bloomFilter.mightContain("14"));
        System.out.println(bloomFilter.mightContain("15"));
        System.out.println(bloomFilter.mightContain("16"));
        System.out.println(bloomFilter.mightContain("17"));
        System.out.println(bloomFilter.mightContain("18"));
        System.out.println(bloomFilter.mightContain("19"));

    }
}
